package com.to8to.kitt;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageCodec;
import io.netty.util.ReferenceCountUtil;

import java.util.List;

import org.apache.thrift.TApplicationException;
import org.apache.thrift.TBase;
import org.apache.thrift.protocol.TMessage;
import org.apache.thrift.protocol.TMessageMeta;
import org.apache.thrift.protocol.TMessageType;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TTransport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.to8to.kitt.esb.ServiceMethodObj;
import com.to8to.kitt.utils.LogUtils;
import com.to8to.kitt.utils.NettyUtils;

public class ThriftMessageCodec extends ByteToMessageCodec<ThriftMessage>{

	public static Logger logger = LoggerFactory.getLogger(ThriftMessageCodec.class);  
	 
	public static int initBuffSize = 64;
	 
	@Override
	protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception 
	{	
		logger.trace("开始解码: " + in);
		
		long t1 = System.currentTimeMillis();
		
		TTransport transport = new TNettyTransport(ctx.channel(), in);
		
		TProtocol protocol = ThriftClassFactory.ProtocolFactory.getProtocol(transport);
		
		ThriftMessage msg = null;
		
		try {
			TMessage head = protocol.readMessageBegin();
			
			logger.trace("解码 head:{}", head);
			
			TMessageMeta meta = new TMessageMeta(head.name);
			
			if(meta != null)
				LogUtils.openTag(meta.tag);
			
			String method = meta.method;
			
			String service = meta.service;
			
			ServiceMethodObj sm = null;
			
			if(service != null)
				sm = ThriftClassFactory.getServiceMethodObj(service);
			
			if(sm == null && ThriftClassFactory.ServiceMethodMap.size() > 0)
				sm = ThriftClassFactory.ServiceMethodMap.values().iterator().next();
			
			if(sm != null)
			{
				byte type = head.type;
				if (type == TMessageType.CALL || type == TMessageType.REPLY) 
				{
					TBase<?,?> body = null;
					if (type == TMessageType.CALL)
					{
						TBase<?,?> tmp = sm.argsMap.get(method);
						if(tmp != null)
							body = tmp.deepCopy();
					}
					else
					{
						TBase<?,?> tmp = sm.resultMap.get(method);
						if(tmp != null)
							body = tmp.deepCopy();
					}
					
					if(body != null)
					{
						body.read(protocol);
						protocol.readMessageEnd();
						msg = new ThriftMessage(head, body);
						
					}
					else 
					{
						logger.error("Unknown method: " + method + ", for service: " + sm.serviceClass.getName());
						TApplicationException x = new TApplicationException(TApplicationException.UNKNOWN_METHOD, "Unknown method: " + method + ", for service: " + sm.serviceClass.getName());
						msg = new ThriftMessage(head, x);
					}
				} 
				else if (type == TMessageType.EXCEPTION)
				{
					TApplicationException x = TApplicationException.read(protocol);
					protocol.readMessageEnd();
					msg = new ThriftMessage(head, x);
				}
				else
				{
					logger.error("Invalid message type: " + type);
					TApplicationException x = new TApplicationException(TApplicationException.INVALID_MESSAGE_TYPE, "Invalid message type: " + type);
					msg = new ThriftMessage(head, x);
				}
			}
			else
			{
				logger.error("Service method not found: " + method);
				TApplicationException x = new TApplicationException(TApplicationException.UNKNOWN_METHOD, "Service method not found: " + method);
				msg = new ThriftMessage(head, x);
			}
			
			msg.setMeta(meta);
			
			out.add(msg);
			
		} catch (Exception th) {
			logger.error("decode failed: " + LogUtils.logStackTrace(th));
//			System.exit(-1);
//			TApplicationException x = new TApplicationException(TApplicationException.INTERNAL_ERROR, "decode failed: " + LogUtils.logStackTrace(th));
//			msg = new ThriftMessage(head, x);
		}
		
		long t2 = System.currentTimeMillis();
		
		logger.trace("解码完成! time:{} ms, size:{} kb, seqid:{}", (t2-t1), (float)in.writerIndex()/1024.0, msg.getSeqid());
		
		logger.trace("消息体:{}", msg.toString());
		
		LogUtils.removeTag();
	}


	@Override
	protected void encode(ChannelHandlerContext ctx, ThriftMessage msg, ByteBuf out) throws Exception 
	{
		logger.trace("开始编码!");
		
		TMessageMeta meta = msg.getMeta();
		
		if(meta == null)
			meta = new TMessageMeta(msg.getHead().name);
		
		LogUtils.openTag(meta.tag);
		
		long t1 = System.currentTimeMillis();
		
		out.capacity(initBuffSize);
		
		TTransport transport = new TNettyTransport(ctx.channel(), out);
		TProtocol protocol = ThriftClassFactory.ProtocolFactory.getProtocol(transport);
		try {
			protocol.writeMessageBegin(msg.getHead());
			if (msg.getHead().type == TMessageType.CALL 
					|| msg.getHead().type == TMessageType.REPLY)
				msg.getBody().write(protocol);
			else
				msg.getException().write(protocol);
			protocol.writeMessageEnd();
			protocol.getTransport().flush();
		} catch (Throwable e) {
			
			NettyUtils.exitIfOutOfMemory(e);
			
			logger.error("Exception:{}\n{}", e, LogUtils.logStackTrace(e));
			
			ReferenceCountUtil.release(out);
		}
		
		long t2 = System.currentTimeMillis();
		
		logger.trace("编码结束! time:{} ms, size:{} kb, seqid:{}", (t2-t1), (float)out.writerIndex()/1024.0, msg.getSeqid());
		
		logger.trace("消息体:{}", msg.toString());
		
		LogUtils.removeTag();
	}

	public static ByteBuf encode(ThriftMessage msg)
	{
		ByteBuf out = Unpooled.buffer(initBuffSize);
		
		TTransport transport = new TNettyTransport(null, out);
		TProtocol protocol = ThriftClassFactory.ProtocolFactory.getProtocol(transport);
		try {
			protocol.writeMessageBegin(msg.getHead());
			if (msg.getHead().type == TMessageType.CALL 
					|| msg.getHead().type == TMessageType.REPLY)
				msg.getBody().write(protocol);
			else
				msg.getException().write(protocol);
			protocol.writeMessageEnd();
			protocol.getTransport().flush();
			return out;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}
	
	
	public static ThriftMessage decode(ByteBuf in) 
	{	
		TTransport transport = new TNettyTransport(null, in);
		
		TProtocol protocol = ThriftClassFactory.ProtocolFactory.getProtocol(transport);
		
		ThriftMessage msg = null;
		
		try {
			TMessage head = protocol.readMessageBegin();
			
			TMessageMeta meta = new TMessageMeta(head.name);
			
			LogUtils.openTag(meta.tag);
			
			String method = meta.method;
			
			String service = meta.service;
			
			ServiceMethodObj sm = ThriftClassFactory.getServiceMethodObj(service);
			
			if(sm == null && ThriftClassFactory.ServiceMethodMap.size() > 0)
				sm = ThriftClassFactory.ServiceMethodMap.values().iterator().next();
			
			byte type = head.type;
			if (type == TMessageType.CALL || type == TMessageType.REPLY) 
			{
				TBase<?,?> body = null;
				if (type == TMessageType.CALL)
				{
					TBase<?,?> tmp = sm.argsMap.get(method);
					if(tmp != null)
						body = tmp.deepCopy();
				}
				else
				{
					TBase<?,?> tmp = sm.resultMap.get(method);
					if(tmp != null)
						body = tmp.deepCopy();
				}
				
				if(body != null)
				{
					body.read(protocol);
					protocol.readMessageEnd();
					msg = new ThriftMessage(head, body);
					
				}
				else 
				{
					logger.error("Unknown method: " + method + ", for service: " + sm.serviceClass.getName());
					TApplicationException x = new TApplicationException(TApplicationException.UNKNOWN_METHOD, "Unknown method: " + method + ", for service: " + sm.serviceClass.getName());
					msg = new ThriftMessage(head, x);
				}
			} 
			else if (type == TMessageType.EXCEPTION)
			{
				TApplicationException x = TApplicationException.read(protocol);
				protocol.readMessageEnd();
				msg = new ThriftMessage(head, x);
			}
			else
			{
				logger.error("Invalid message type: " + type);
				TApplicationException x = new TApplicationException(TApplicationException.INVALID_MESSAGE_TYPE, "Invalid message type: " + type);
				msg = new ThriftMessage(head, x);
			}
			
			msg.setMeta(meta);
			
		} catch (Exception e) {
			logger.error("", e);
		}
		
		return msg;
	}
	
	public static void decode(byte[] bytes, TBase<?,?> tbase)
	{
		ByteBuf in = Unpooled.wrappedBuffer(bytes);
		
		TTransport transport = new TNettyTransport(null, in);
		
		TProtocol protocol = ThriftClassFactory.ProtocolFactory.getProtocol(transport);
		
		try {
			
			tbase.read(protocol);
			
		} catch (Exception e) {
			logger.error("Exception:{}\n{}", e, LogUtils.logStackTrace(e));
		}
	}
	
	public static byte[] encode(TBase<?,?> tbase)
	{
		ByteBuf out = Unpooled.buffer(initBuffSize);
		
		TTransport transport = new TNettyTransport(null, out);
		
		TProtocol protocol = ThriftClassFactory.ProtocolFactory.getProtocol(transport);
		
		try {
			
			tbase.write(protocol);
			
			protocol.getTransport().flush();
			
			return out.array();
			
		} catch (Exception e) {
			out.clear();
			logger.error("Exception:{}\n{}", e, LogUtils.logStackTrace(e));
		}
		
		return null;
	}
	
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    	NettyUtils.exitIfOutOfMemory(cause);
        ctx.fireExceptionCaught(cause);
    }
	
	public static void main(String[] args)
	{
		
	}
}
